#include "myRPCProvider.h"

#include "rpcHeader.pb.h"

#include <arpa/inet.h>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <fstream>
#include <functional>
#include <glog/logging.h>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
#include <google/protobuf/message.h>
#include <google/protobuf/service.h>
#include <google/protobuf/stubs/callback.h>
#include <ifaddrs.h>
#include <memory>
#include <muduo/net/Callbacks.h>
#include <muduo/net/InetAddress.h>
#include <muduo/net/TcpServer.h>
#include <netdb.h>
#include <netinet/in.h>
#include <string>
#include <sys/socket.h>
#include <unistd.h>

namespace RPC
{

    void MyRpcProvider::NotifyService( google::protobuf::Service *service )
    {
        // 创建存储服务信息的结构体
        ServiceInfo service_info;

        // 获取服务的服务描述符
        const google::protobuf::ServiceDescriptor *sdpr = service->GetDescriptor();

        // 获取服务名
        std::string service_name = sdpr->name();
        // 获取服务中方法的个数
        int methodCount = sdpr->method_count();

        // 需要根据方法的个数去获得方法的名称
        service_info.m_servie = service;
        for ( int i = 0; i < methodCount; ++i )
        {
            const google::protobuf::MethodDescriptor *mdpr = sdpr->method( i );
            // 通过方法描述符对象来获得方法名
            std::string method_name = mdpr->name();
            service_info.m_methodMap.emplace( method_name, mdpr );
        }

        // 外部还需要将服务保存在存储多种服务的容器中，上述的循环只是将一种服务中的方法属性填充到一个服务的容器中
        m_service_map.emplace( service_name, service_info );
    }

    // 启动rpc服务节点，开始提供rpc远程网络调用服务
    void MyRpcProvider::Run( int nodeIndex, short port )
    {
        struct ifaddrs *ifaddr, *ifa;
        char ip[ INET6_ADDRSTRLEN ];

        // 这是获取网络主机中的所有接口的
        if ( getifaddrs( &ifaddr ) == -1 )
        {
            perror( "getifaddrs" );
            LOG( ERROR ) << "获取本地网络接口失败";
            return;
        }

        // 遍历每个接口
        for ( ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next )
        {
            if ( ifa->ifa_addr == NULL )
                continue;

            int family = ifa->ifa_addr->sa_family;

            if ( family == AF_INET )
            {
                std::string interface_name( ifa->ifa_name );
                if ( interface_name == "lo" )
                    continue;
                struct sockaddr_in *addr = reinterpret_cast< struct sockaddr_in * >( ifa->ifa_addr );
                inet_ntop( AF_INET, &addr->sin_addr, ip, sizeof( ip ) );
            }
            // 目前不处理ipv6的地址
            //  else if ( family == AF_INET6 )
            //  {
            //      struct sockaddr_in6 *addr6 = reinterpret_cast< struct sockaddr_in6 * >( ifa->ifa_addr );
            //      inet_ntop( AF_INET6, &addr6->sin6_addr, ip, sizeof( ip ) );
            //  }
        }

        freeifaddrs( ifaddr );
        std::string ip_str( ip );

        // 将ip写入到文件中去
        std::string node = "node" + std::to_string( nodeIndex );
        std::ofstream outFile( "test.conf", std::ios::app );
        if ( !outFile.is_open() )
        {
            perror( "打开文件失败" );
            LOG( ERROR ) << "打开文件失败";
            exit( EXIT_FAILURE );
        }

        outFile << node + "ip=" + ip << "\n";
        outFile << node + "port=" + std::to_string( port ) << "\n";
        outFile.close();

        // 1、初始化muduo的网络地址和端口
        muduo::net::InetAddress address( ip, port );

        // 2、使用地址和EVentLoop创建muduo的TCPServer对象
        m_muduo_server = std::make_shared< muduo::net::TcpServer >( &m_eventLoop, address, "MyRpcProvider" );

        // 3、绑定连接发生时的回调函数
        m_muduo_server->setConnectionCallback( std::bind( &MyRpcProvider::OnConnection, this, std::placeholders::_1 ) );
        m_muduo_server->setMessageCallback(
            std::bind( &MyRpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3 ) );

        // 4、设置muduo库的线程数
        m_muduo_server->setThreadNum( 5 );

        // rpc服务端准备启动，打印信息
        std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

        // 5、启动muduo服务端
        m_muduo_server->start();
        m_eventLoop.loop();
    }

    void MyRpcProvider::OnConnection( const std::shared_ptr< muduo::net::TcpConnection > &conn )
    {
        // 这是一个连接回调函数，当 TCP 连接状态发生变化时（例如新连接建立或连接关闭）会被调用。
        // 通过检查连接的状态（conn->connected()），你可以决定如何处理这个连接。
        if ( conn->connected() == false )
        {
            conn->shutdown();
        }
    }

    /**
     * @brief 在框架内部，RpcProvider和RpcConsumer协商好之间通信用的protobuf数据类型
                service_name method_name args    定义proto的message类型，进行数据头的序列化和反序列化
                                                service_name method_name args_size
                16UserServiceLoginzhang san123456

                header_size(4个字节) + header_str + args_str
                10 "10"
                10000 "1000000"
                std::string   insert和copy方法
     *          这个函数是：每次客户端都发送请求后，执行的函数，通过std::bind和muduo库进行绑定；
     * @param conn
     */
    void MyRpcProvider::OnMessage( const std::shared_ptr< muduo::net::TcpConnection > &conn, muduo::net::Buffer *buffer, muduo::Timestamp )
    {

        // 从Buffer中读取数据使用retrieveAllAsString()
        std::string data = buffer->retrieveAllAsString();

        // 因为我们是使用变长编码和protobuf提供的字符流的方式写入的数据，我们也需要使用一样的方式来处理数据
        // 输出我们使用的是codeoutput流而输入我们则是要使用codeintput流，但是这个流是高级流需要使用较低级别的作为载体
        google::protobuf::io::ArrayInputStream arrayInputStream( data.data(), data.size() );
        google::protobuf::io::CodedInputStream codeInput( &arrayInputStream );

        // 从数据中获得rpcHeader的长度，该长度是写入到了发送的数据里面的（即我们接收到的数据里面）
        uint32_t header_size{};
        codeInput.ReadVarint32( &header_size );

        // 安全读取，设置读取限制,使用PushLimit
        /**
         * @brief PushLimit方法，是把限制设置到CodedInputStream中并返回之前的流的限制，这样做可以当安全读取完成后，设置会之前的读取限制；
         *
         */
        google::protobuf::io::CodedInputStream::Limit pre_limit = codeInput.PushLimit( header_size );

        // 读取数据
        std::string rpc_header_str;
        codeInput.ReadString( &rpc_header_str, header_size );
        // 回复之前的限制
        codeInput.PopLimit( pre_limit );

        // 将rpc_header_str进行反序列化
        RPCHeader rpcHeader;
        if ( !rpcHeader.ParseFromString( rpc_header_str ) )
        {
            // 数据头反序列化失败
            std::cout << "rpc_header_str:" << rpc_header_str << " parse error!" << std::endl;
            LOG( ERROR ) << "rpc_header_str:" << rpc_header_str << " parse error!";
            return;
        }

        std::string service_name = rpcHeader.service_name();
        std::string method_name = rpcHeader.method_name();
        uint32_t args_size = rpcHeader.args_size();

        // 我们需要获得参数数据
        // codeInput的ReadString方法在读取完数据后，会标记前面的数据已经读取了

        std::string args_str;
        bool read_args_success = codeInput.ReadString( &args_str, args_size );
        if ( !read_args_success )
        {
            LOG( ERROR ) << "readString is error!";
            return;
        }

        // 去map中查找服务的信息
        auto it = m_service_map.find( service_name );
        if ( it == m_service_map.end() )
        {
            std::cout << "服务：" << service_name << " is not exist!" << std::endl;
            LOG( ERROR ) << "服务：" << service_name << " is not exist!";
            std::cout << "当前已经有的服务列表为:";
            for ( auto item : m_service_map )
            {
                std::cout << item.first << " ";
            }
            std::cout << std::endl;
            return;
        }

        // 然后去找方法名
        ServiceInfo service_info = it->second;
        auto mit = service_info.m_methodMap.find( method_name );
        if ( mit == service_info.m_methodMap.end() )
        {
            std::cout << service_name << ":" << method_name << " is not exist!" << std::endl;
            LOG( ERROR ) << service_name << ":" << method_name << " is not exist!";
            return;
        }

        // 找到了的话就获取服务的描述符
        google::protobuf::Service *service = service_info.m_servie;

        // 然后获取方法的描述符
        const google::protobuf::MethodDescriptor *method = mit->second;

        // 获取Message对象，这是动态获取与某个RPC方法对应的请求消息类型
        google::protobuf::Message *request = service->GetRequestPrototype( method ).New();

        // 将参数序列化成message对象
        if ( !request->ParseFromString( args_str ) )
        {
            std::cerr << "request parse error, content: " << args_str << std::endl;
            LOG( ERROR ) << "request parse error, content:" << args_str;
            return;
        }

        // 然后就是生成响应了
        google::protobuf::Message *response = service->GetResponsePrototype( method ).New();

        // 设置RPC的回调函数，当CallMethod执行的调用函数执行完毕之后，就会执行这个函数；回调函数
        google::protobuf::Closure *done =
            google::protobuf::NewCallback< MyRpcProvider, const muduo::net::TcpConnectionPtr &, google::protobuf::Message * >(
                this, &MyRpcProvider::SendRpcResponse, conn, response );

        // 这里面调用的是我们在protoc里面使用service关键字写的服务，使用protoc生成的.pb.h和.pb.cc文件中的服务类的CallMethod方法；
        // 这个方法会根据method的方法描述符去找相应的方法，去调用，但是讷这个相应的方法在这个文件中是虚函数所以我们需要定义我们自己的类去继承生产的这个基类，去重写这些需要远程调用的
        // 方法，这样才能完成远程过程调用
        // 注意不要和myRPCChannel类的CallMethod混淆，它的作用是客户端stub发送远程调用请求的底层实现；
        service->CallMethod( method, nullptr, request, response, done );
        LOG( INFO ) << "RPC远程过程调用开始调用" << method->name() << "方法";
        /*
            CallMethod的参数解释：
                1、方法的描述符，每个定义在服务里面的方法都会有一个唯一的方法描述符;
                2、控制器，用于控制错误处理；
                3、请求的方法的参数，比如本案例中的任期和Index;
                4、回调响应，通常是将调用的函数产生的数据发送给请求方，在此之前会将数据序列化成message消息，这个response会在对调函数中使用
                5、回调函数，需要将产生的响应发送给请求方
        */
    }
    

    // 我们需要设置当处理完毕请求的调用函数之后，需要将数据传递给请求方，此时就需要使用这个了response的回调函数，执行完回调函数后，处理发送
    void MyRpcProvider::SendRpcResponse( const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response )
    {
        // 首先我们需要使用序列化，将response对象序列化成字符串,这里是发送处理的响应的
        std::string response_str;
        if ( response->SerializePartialToString( &response_str ) )
        {
            conn->send( response_str );
            LOG( INFO ) << "远程过程调用执行完毕，数据发送成功";
        }
        else
        {
            std::cerr << "send message is error" << std::endl;
            LOG( ERROR ) << "send message is error";
        }
    }

    /**
     * @brief 析构函数，需要让服务器停止了
     *
     */

    MyRpcProvider::~MyRpcProvider()
    {
        std::cout << "[func - MyRpcProvider::~MyRpcProvider()]: ip和port信息：" << m_muduo_server->ipPort() << std::endl;
        LOG( INFO ) << "[服务器停止运行]: 服务器ip和port信息：" << m_muduo_server->ipPort();
        m_eventLoop.quit();
    }
}  // namespace RPC